/*
 * ===========================================================================
 *
 *      Filename: io_base.c
 *   Description: implement network io operation depend on epoll
 *        Author: Ye wenhao, wenhaoye@126.com
 *  Organization: 
 *
 * ===========================================================================
 */
#include "def.h"
#include "io_tools.h"
#include "epoll_fd.h"
#include "list.h"
#include "buffer.h"
#include "io_base.h"
#include <pthread.h>

#define IO_BASE_CTRL_UNIX_PATH      "/tmp/io-ctrl"

#define IO_BASE_STATE_RUNNING   1
#define IO_BASE_STATE_EXIT      0

struct io_base {
    io_base_conf_t *conf;
    uint8_t state;
    
    sock_t sock;
    sock_t io_ctrl;
    io_ctrl_msg_t ctrl_msg;

	int sig_fd;
	int signo;
	sigset_t mask;
    
    epoll_fd_t efd;
    uint16_t client_nums;
    struct dl_list clients;
    
    io_event_t io_ev;
    struct dl_list io_evs;
};

static void io_event_term(io_event_t *self);
static int io_client_recv(io_client_t *self);
static int io_client_send(io_client_t *self);

/*****************************************************************************
 *                    io base wraper
 ****************************************************************************/
static inline int io_base_sock_fd(io_base_t *self)
{
    return sock_fd(&self->sock);
}

static inline int io_base_ctrl_fd(io_base_t *self)
{
    return sock_fd(&self->io_ctrl);
}

static inline int io_base_epoll_add(io_base_t *self, int fd, int event, void *ptr)
{
    return epoll_fd_add(&self->efd, fd, event, ptr);
}

static inline uint8_t io_base_state(io_base_t *self)
{
	return self->state;
}

static inline void io_base_set_state(io_base_t *self, uint8_t state)
{
	self->state = state;
}

/*****************************************************************************
 *                             io base
 ****************************************************************************/
io_base_t *io_base_new(io_base_conf_t *conf)
{   
    io_base_t *self = (io_base_t *)MALLOC(sizeof(io_base_t));
    if (self) {
        char info[256] = {0};
        snprintf(info, sizeof(info), "@tcp:%s:%u", conf->ip, conf->port);
        if (sock_init(&self->sock, info) < 0)
            goto _fail;
        
        snprintf(info, sizeof(info), "@unix:%s-%s", IO_BASE_CTRL_UNIX_PATH, conf->name);
        if (sock_init(&self->io_ctrl, info) < 0)
            goto _fail;
        
		sigemptyset(&self->mask);
		if (conf->mask) {
			memcpy(&self->mask, conf->mask, sizeof(sigset_t));	
		}
		sigaddset(&self->mask, SIGINT);
		sigaddset(&self->mask, SIGTERM);
		sigaddset(&self->mask, SIGQUIT);

		self->sig_fd = signalfd(-1, &self->mask, SFD_NONBLOCK | SFD_CLOEXEC);
		if (self->sig_fd < 0) {
			ERR("signalfd fail");	
			goto _fail;
		}

        dl_list_init(&self->clients);
        dl_list_init(&self->io_evs);
		io_event_term(&self->io_ev);
        
        if (epoll_fd_create(&self->efd, 0) < 0) {
            ERR("epoll_fd_create fail");
            goto _fail;
        }
        
        if (io_base_epoll_add(self, io_base_sock_fd(self), EPOLLIN, self) < 0)
            goto _fail;

        if (io_base_epoll_add(self, io_base_ctrl_fd(self), EPOLLIN, NULL) < 0)
            goto _fail;

		if (io_base_epoll_add(self, self->sig_fd, EPOLLIN, NULL) < 0)
			goto _fail;

		io_base_set_state(self, IO_BASE_STATE_RUNNING);
    }
    
    return self;

_fail:
    io_base_destroy(&self);
    return NULL;
}

void io_base_destroy(io_base_t **p_self)
{
    if (p_self && *p_self) {
        io_base_t *self = *p_self;
		io_base_set_state(self, IO_BASE_STATE_EXIT);
        epoll_fd_release(&self->efd);
        self->client_nums = 0;
        dl_list_init(&self->io_evs);
        dl_list_init(&self->clients);
        sock_term(&self->io_ctrl);
        sock_term(&self->sock);
		safe_close(self->sig_fd);
        safe_free(*p_self);
    }
}

int io_base_accept(io_base_t *self, sock_t *s)
{
    return sock_accept(&self->sock, s);
}

int io_base_signo(io_base_t *self)
{
	return self->signo;
}

static int io_base_add_client(io_base_t *self, io_client_t *cli)
{
	ENTER();
    cli->event = EPOLLIN;
    if (epoll_fd_add(&self->efd, sock_fd(&cli->sock), cli->event, cli) < 0)
        return -1;
    
    dl_list_add_tail(&self->clients, &cli->head);
    self->client_nums++;
    cli->base = self;
    return 0;
}

static int io_base_del_client(io_base_t *self, io_client_t *cli)
{
	ENTER();
    epoll_fd_del(&self->efd, sock_fd(&cli->sock));
    dl_list_del(&cli->head);
    self->client_nums--;
    cli->base = NULL;
    return 0;
}

int io_base_stop(io_base_t *self)
{
	ENTER();
    io_ctrl_msg_t msg;
    msg.sync = IO_BASE_MSG_SYNC;
    msg.id = IO_BASE_MSG_ID_EXIT;

    char path[256] = {0};
    snprintf(path, sizeof(path), "%s-%s", IO_BASE_CTRL_UNIX_PATH, self->conf->name);
    int sock = unix_udp_client_create(path);
    if (sock < 0) {
        ERR("unix_udp_client_create fail");
        return -1;
    }

	if (send(sock, &msg, sizeof(msg), 0) < 0) {
		ERR("send fail");
		return -1;
	}

#if 0
	while (io_base_state(self) != IO_BASE_STATE_EXIT) {
		DBG("waiting...");
		sleep (1);
	}
#endif
	return 0;
}

/****************************************************************************
 *                     event handler
 ***************************************************************************/
static int is_io_event_active(io_event_t *self)
{
	if (dl_list_empty(&self->head))
        return 0;

    if (self->head.next == NULL && self->head.prev == NULL)
        return 0;

    return 1;
}

static void io_event_init(io_event_t *self, struct dl_list *l,
                                uint16_t event, void *ctx)
{
    self->event = event;
    self->ctx = ctx;
	if (is_io_event_active(self))
		return ;
	//DBG("dl_list_add_tail");
    dl_list_add_tail(l, &self->head);
}

static void io_event_term(io_event_t *self)
{
    //dl_list_del(&self->head);
    dl_list_init(&self->head);
    self->event = IO_EVENT_UNKNOW;
    self->ctx = NULL;
}

static void handle_signal_event(io_base_t *base)
{
	struct signalfd_siginfo fdsi;
	memset(&fdsi, 0, sizeof(fdsi));
	ssize_t len = read(base->sig_fd, &fdsi, sizeof(fdsi));
	if (len < sizeof(fdsi)) {
		ERR("read signalfd fail");
		return ;
	}

	base->signo = fdsi.ssi_signo;
	uint16_t event = base->io_ev.event | IO_EVENT_SIGNAL;
	return io_event_init(&base->io_ev, &base->io_evs, event, base);
}

static void handle_accept_event(io_base_t *base)
{
	uint16_t event = base->io_ev.event | IO_EVENT_ACCEPT;
	return io_event_init(&base->io_ev, &base->io_evs, event, base);
}

static void handle_err_event(io_client_t *cli)
{
	uint16_t event = cli->io_ev.event | IO_EVENT_ERROR;
	return io_event_init(&cli->io_ev, &cli->base->io_evs, event, cli);
}

static int handle_ctrl_event(io_base_t *self)
{
    if (recvfrom(sock_fd(&self->io_ctrl), &self->ctrl_msg, sizeof(self->ctrl_msg),
                0, NULL, NULL) < 0) {
        ERR("recvfrom fail");
        return -1;
    }

    if (self->ctrl_msg.sync != IO_BASE_MSG_SYNC) {
        DBG("invlaid sync(%u)", self->ctrl_msg.sync);
        return -1;
    }

    switch (self->ctrl_msg.id) {
    case IO_BASE_MSG_ID_EXIT:
		io_base_set_state(self, IO_BASE_STATE_EXIT);
        break;
    default:
        break;
    }

	uint16_t event = self->io_ev.event | IO_EVENT_CTRL;
	io_event_init(&self->io_ev, &self->io_evs, event, self);
    return 0;
}

static void handle_in_event(io_client_t *cli)
{
	uint16_t event = cli->io_ev.event;
    int ret = io_client_recv(cli);
    if (ret < 0) {
        ERR("in(%d) fail", sock_fd(&cli->sock));
		event |= IO_EVENT_ERROR;
        return io_event_init(&cli->io_ev, &cli->base->io_evs, event, cli);
    }

    if (ret == 0) {
        DBG("in(%d) has closed", sock_fd(&cli->sock));
		event |= IO_EVENT_CLOSE;
        return io_event_init(&cli->io_ev, &cli->base->io_evs, event, cli);
    }

	event |= IO_EVENT_READ;
    return io_event_init(&cli->io_ev, &cli->base->io_evs, event, cli);
}

static void handle_out_event(io_client_t *cli)
{
	uint16_t event = cli->io_ev.event;
    if (io_client_send(cli) < 0) {
        ERR("io_client_send fail");
		event |= IO_EVENT_ERROR;
        return io_event_init(&cli->io_ev, &cli->base->io_evs, event, cli);
    }

	if (buffer_len(&cli->out_buf))
		return ;

    cli->event &= (~EPOLLOUT);
    epoll_fd_mod(&cli->base->efd, sock_fd(&cli->sock), cli->event, cli);
	event |= IO_EVENT_WRITE;
	return io_event_init(&cli->io_ev, &cli->base->io_evs, event, cli);
}

int io_base_loop(io_base_t *self, int timeout)
{
    if (io_base_state(self) != IO_BASE_STATE_RUNNING)
        return -1;

    int nums = epoll_fd_wait(&self->efd, timeout);
	//DBG("event nums=%d", nums);
    if (nums < 0) {
        ERR("epoll_fd_wait fail");
		io_base_set_state(self, IO_BASE_STATE_EXIT);
        return nums;
    }

    if (nums == 0)
        return nums;

    struct epoll_event *events = epoll_events(&self->efd);
    dl_list_init(&self->io_evs);
    int i;
    for (i = 0; i < nums; ++i) {
		if (io_base_state(self) != IO_BASE_STATE_RUNNING)
			break;

        if (events[i].data.ptr == self) {
			//DBG("accept event");
            handle_accept_event(self);
            continue;
        }

		if (events[i].data.fd == self->sig_fd) {
			DBG("signal event");
			handle_signal_event(self);
			continue;
		}

        if (events[i].data.fd == io_base_ctrl_fd(self)) {
			DBG("ctrl event");
            handle_ctrl_event(self);
            continue;
        }

        if (events[i].events & EPOLLERR) {
			DBG("error event");
            handle_err_event((io_client_t *)(events[i].data.ptr));
            continue;
        }

        if (events[i].events & EPOLLIN) {
			//DBG("in event");
            handle_in_event((io_client_t *)(events[i].data.ptr));
        }

        if (events[i].events & EPOLLOUT) {
			//DBG("out event");
            handle_out_event((io_client_t *)(events[i].data.ptr));
        }
    }

    return nums;
}


uint32_t io_base_ev_nums(io_base_t *self)
{
    return dl_list_len(&self->io_evs);
}

void io_event_iterator_init(io_base_t *base, io_event_iterator_t *it)
{
	it->base = base;
	dl_list_add(&base->io_evs, &it->evs);
	dl_list_del(&base->io_evs);
	dl_list_init(&base->io_evs);
}

io_event_t *io_events_next(io_event_iterator_t *it)
{
	if (io_base_state(it->base) != IO_BASE_STATE_RUNNING) {
		DBG("io_base is not running state!");	
		return NULL;
	}

	if (dl_list_empty(&it->evs)) {
		//DBG("empty");
		return NULL;
	}
    
	//DBG("nums: %u", dl_list_len(&it->evs));
	io_event_t *io_ev = dl_list_entry(it->evs.next, io_event_t, head);
	dl_list_del(&io_ev->head);

    memcpy(&it->io_ev, io_ev, sizeof(io_event_t));
    io_event_term(io_ev);
	return &it->io_ev;
}

void io_event_iterator_term(io_event_iterator_t *it)
{
	dl_list_init(&it->evs);
	it->base = NULL;
    io_event_term(&it->io_ev);
}

struct dl_list *io_base_ev_list(io_base_t *self)
{
    return &self->io_evs;
}

/*****************************************************************************
 *                           io client
 ****************************************************************************/
int io_client_init(io_client_t *cli, sock_t *s, io_base_t *base)
{
	ENTER();
    if (cli == NULL || s == NULL || base == NULL) {
        DBG("NULL point");
        return -1;
    }
    
    if (buffer_init(&cli->out_buf, 512) < 0)
        return -1;

    cli->base = base;
    cli->in_len = 0;
    //cli->event = EPOLLIN;
	if (s != &cli->sock)
		memcpy(&cli->sock, s, sizeof(*s));
    cli->ref = 0;
	io_event_term(&cli->io_ev);

    if (io_base_add_client(base, cli) < 0) {
        ERR("io_base_add_client fail");
        goto _fail;
    }
    
	cli->ref++;
    return 0;

_fail:
    buffer_term(&cli->out_buf);
    return -1;
}

static void _io_client_term(io_client_t *cli)
{
	ENTER();
    io_client_send(cli);
    sock_term(&cli->sock);
    //io_base_del_client(cli->base, cli);
    buffer_term(&cli->out_buf);
}

uint8_t io_client_put(io_client_t *self)
{
    self->ref--;
    if (!self->ref) {
		io_base_del_client(self->base, self);
        _io_client_term(self);
        return 0;
    }

    return self->ref;
}

void io_client_term(io_client_t *cli)
{
    io_client_put(cli);
}

static int io_client_recv(io_client_t *self)
{
    int ret;
    ret = fd_read(sock_fd(&self->sock), self->in_buf, sizeof(self->in_buf));
    if (ret >= 0)
        self->in_len = ret;

    return ret;
}

static int io_client_send(io_client_t *self)
{
    size_t len = buffer_len(&self->out_buf);
	if (len > 0)
		return buffer_to_fd(&self->out_buf, sock_fd(&self->sock), len);
	return 0;
}

int io_client_write(io_client_t *self, void *data, size_t len)
{
	int tmp_event = self->event;
	if (data == NULL || len == 0)
		goto _out;

    if (buffer_add(&self->out_buf, data, len) < 0)
        return -1;

_out:
    if (buffer_len(&self->out_buf)) {
		if ((self->event & EPOLLOUT) == 0)
			self->event |= EPOLLOUT;
    } else {
		if (self->event & EPOLLOUT)
			self->event &= (~EPOLLOUT);
	}

	if (tmp_event == self->event)
		return 0;

	return epoll_fd_mod(&self->base->efd, sock_fd(&self->sock), self->event, self);
}
